package com.mango.ch09;

import java.io.IOException;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFlatMapFunction;

import com.mango.HDFSTools.HDFSAPI;

import scala.Tuple2;

public class SparkFriendRecommendJob {
	static Path inputPath = new Path("/SparkFriendRecommendJob/input");// input
	static Path outPath = new Path("/SparkFriendRecommendJob/output");

	@SuppressWarnings({ "serial", "unused", "resource" })
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		// 创建输入文件夹
		Configuration conf = new Configuration();
		try {
			HDFSAPI hdfs = new HDFSAPI(conf);
			// hdfs.createDirectory(transactions);
			// 先删除已经有的输出文件夹
			hdfs.delDirectory(outPath);
			hdfs.closeFilseSystem();
			// hdfs.orpOver();
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
			System.out.println("----------文件操作失败");
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (URISyntaxException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		JavaSparkContext jsc = new JavaSparkContext();
		// 创建文本RDD
		JavaRDD<String> lines = jsc.textFile(inputPath.toString());
		// debug0
		List<String> debug1 = lines.collect();
		for (String string : debug1) {
			System.out.println("debug1 record=" + string);
		}
		// 创建业务键值对
		JavaPairRDD<Long, Tuple2<Long, Long>> pairs = lines
				.flatMapToPair(new PairFlatMapFunction<String, Long, Tuple2<Long, Long>>() {

					@Override
					public Iterator<Tuple2<Long, Tuple2<Long, Long>>> call(String t) throws Exception {
						// TODO Auto-generated method stub
						String[] tokens = t.split(" ");
						Long person = Long.valueOf(tokens[0]);
						String friendString = tokens[1];
						String[] friendsTokeized = friendString.split(",");
						List<Long> friends = new ArrayList<>();
						List<Tuple2<Long, Tuple2<Long, Long>>> mapperout = new ArrayList<Tuple2<Long, Tuple2<Long, Long>>>();
						// 输入所有直接好友
						System.out.println("输出直接好友列表" + friendString + " to " + friendsTokeized.toString());
						for (String fid : friendsTokeized) {
							System.out.println("处理直接好友" + fid);
							Long toUser = Long.valueOf(fid);
							friends.add(toUser);
							Tuple2<Long, Long> directFriend = new Tuple2<Long, Long>(toUser, (long) -1);// 设置当前用户与好友的直接关系
							mapperout.add(T2(person, directFriend));
						}
						// 输出所有可能的好友，即所有推荐情况
						for (int i = 0; i < friendsTokeized.length; i++)
							for (int j = i + 1; j < friendsTokeized.length; j++) {
								// 可能的好友1
								Tuple2<Long, Long> pf1 = T2(friends.get(i), person);
								mapperout.add(T2(friends.get(j), pf1));
								Tuple2<Long, Long> pf2 = T2(friends.get(j), person);
								mapperout.add(T2(friends.get(i), pf2));
								// 可能的好友2
							}
						return mapperout.iterator();
					}

				});
		// debug2
		List<Tuple2<Long, Tuple2<Long, Long>>> debug2 = pairs.collect();
		for (Tuple2<Long, Tuple2<Long, Long>> t : debug2) {
			System.out.print("debug2 user:" + t._1 + "被推荐朋友" + t._2._1 + " 共同好友是" + t._2._2);
		}

		// 规约数据，找出正真的可推荐好友
		JavaPairRDD<Long, Iterable<Tuple2<Long, Long>>> grouped = pairs.groupByKey();
		JavaPairRDD<Long, String> result = grouped.mapValues(new Function<Iterable<Tuple2<Long, Long>>, String>() {

			@Override
			public String call(Iterable<Tuple2<Long, Long>> v1) throws Exception {
				// TODO Auto-generated method stub
				Map<Long, List<Long>> mutualFriends = new HashMap<>();
				for (Tuple2<Long, Long> t : v1) {
					Long toUser = t._1;
					Long mutualFriend = t._2;
					boolean isFriend = (mutualFriend == -1);
					if (mutualFriends.containsKey(toUser)) {
						if (isFriend)
							// 如果该推荐人和当前用户已经是直接好友，则添加一个空的列表进共同好友列表
							// 后面处理该集合时，遇到某key的value是个null的列表，则表示该推荐人与当前用户是直接好友
							mutualFriends.put(toUser, null);
						else if (mutualFriends.get(toUser) != null) {
							mutualFriends.get(toUser).add(mutualFriend);// 否则将共同好友记录到列表中
						}
					} else {
						// map中还没有该key的记录
						// 但已经是直接好友，则添加一个空的列表进共同好友列表
						if (isFriend)
							mutualFriends.put(toUser, null);
						else {
							List<Long> list = new ArrayList<>();
							list.add(mutualFriend);
							mutualFriends.put(toUser, list);
						}
					}
				}

				return buildRecommendations(mutualFriends);
			}
		});
		result.saveAsTextFile(outPath.toString() + "/result");
		// 输出结果 debug3
		List<Tuple2<Long, String>> debug3 = result.collect();
		for (Tuple2<Long, String> tuple2 : debug3) {
			System.out.println("用户" + tuple2._1 + " 推荐人列表：" + tuple2._2);
		}
	}

	public static String buildRecommendations(Map<Long, List<Long>> mutualFriends) {
		StringBuilder recommentions = new StringBuilder();
		for (Entry<Long, List<Long>> entry : mutualFriends.entrySet()) {
			if (entry.getValue() == null)
				// 已经是直接好友不用推荐
				continue;
			recommentions.append("推荐" + entry.getKey());
			recommentions.append("(");
			recommentions.append(entry.getValue().size() + "个共同好友");
			recommentions.append(":+共同好友列表");
			recommentions.append(entry.getValue());
			recommentions.append(") ");
		}
		return recommentions.toString();
	}

	static Tuple2<Long, Long> T2(Long a, Long b) {
		return new Tuple2<Long, Long>(a, b);
	}

	static Tuple2<Long, Tuple2<Long, Long>> T2(Long a, Tuple2<Long, Long> b) {
		return new Tuple2<Long, Tuple2<Long, Long>>(a, b);
	}
}
